package ru.mipt.victator.pagerank.finalization;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;

public class SequenceFilesCombiner
{
	private static void _append(FileSystem FS, Configuration conf, Path in, SequenceFile.Writer out) throws IOException
	{
		SequenceFile.Reader reader = new SequenceFile.Reader(FS,in,conf);
		boolean read = true;
		Text text = new Text();
		FloatWritable floatw = new FloatWritable();
		while(read)
		{
			read = reader.next(text, floatw);
			if (read)
			{
				out.append(text, floatw);
			}
		}
		reader.close();
	}
	
	public static void combine(FileSystem FS, Configuration conf,Path ins, Path out) throws IOException
	{
		SequenceFile.Writer writer = SequenceFile.createWriter(	FS,
				conf,
				out,
				Text.class,
				FloatWritable.class
				);

		if (FS.isFile(ins))
		{
			_append(FS,conf,ins,writer);
		}
		else
		{
			Path[] children = FS.listPaths(ins);
			for (Path child : children)
			{
				_append(FS, conf, child,writer);
			}
		}
		writer.close();
	}
}
